package com.mars.rabbitmq.delay;

import com.mars.rabbitmq.normal.RabbitMqConfig;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.GetResponse;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.concurrent.*;

/**
 * @Author King
 * @create 2020/8/19 14:21
 */
@Slf4j
@Component
public class KingOrderConsumer {

    /**
     * push模式
     *
     * @param message
     * @param channel
     */
    @RabbitListener(queues = OrderRabbitMqConfig.QUEUE)
    public void onMessage(Message message, Channel channel) {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //TODO 处理业务
            log.info("接收订单信息:" + new String(message.getBody()));
            log.info("先查询是否已经支付");
            log.info("超时支付取消订单");
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        try {
            //TODO 手动确认消息有消费，放在业务逻辑之后
            channel.basicAck(deliveryTag, true);
        } catch (IOException e) {
            //TODO 手动确认失败逻辑
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 主动拉取的模式 pull
     */
    public void onPull() {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("dev.rabbitmq.gms.vvtechnology.cn");
        factory.setPort(5672);
        factory.setPassword("vvoaPassword");
        factory.setUsername("vvoaUsername");

        Connection connection = null;
        try {
            connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(RabbitMqConfig.QUEUE, true, false, false, null);
            ExecutorService executorService = new ThreadPoolExecutor(1, 1,
                    0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        GetResponse response = null;
                        try {
                            response = channel.basicGet(RabbitMqConfig.QUEUE, true);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                        try {
                            if (null != response) {
                                log.info(new String(response.getBody(), "UTF-8"));
                            }
                        } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });

        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }


}
